package com.bds.btcdc.adapter.okex.market;

import com.alibaba.fastjson.JSONObject;
import com.bds.btcdc.model.CoinPairExchanges;
import com.bds.btcdc.model.MarketData;
import com.bds.btcdc.api.model.Result;
import com.bds.btcdc.util.Const;
import com.bds.btcdc.util.EhcacheUtils;
import com.bds.btcdc.util.Utils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.simp.stomp.StompSession;

import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

/**
 * Okex的WebSocket客戶端，接受okex的行情更新推送
 */
public class OkexReceiver extends WebSocketClient {
	private Logger logger = LoggerFactory.getLogger(getClass());

	StompSession session;
	List<CoinPairExchanges> okex_pairs;

	public OkexReceiver(StompSession session, URI serverURI, List<CoinPairExchanges> okex_pairs) {
		super(serverURI);
		this.session = session;
		this.okex_pairs = okex_pairs;
	}

	public OkexReceiver(URI serverUri, Draft draft) {
		super(serverUri, draft);
	}

	public OkexReceiver(URI serverURI) {
		super(serverURI);
	}

	@Override
	public void onOpen(ServerHandshake handshakedata) {
		subMarketInfo();
		logger.info("######### Okex 连接成功 #############");
	}

	/**
	 * 订阅行情
	 */
	private void subMarketInfo(){
		String dataMsg = "";
		for (int i = 0; i < okex_pairs.size(); i++) {
			String pair = okex_pairs.get(i).getPair_code();
			dataMsg += "{'event':'addChannel','channel':'ok_sub_spot_" + pair + "_ticker'},";
		}
		dataMsg.substring(0, dataMsg.length() - 1);
		dataMsg = "[" + dataMsg + "]";
		// System.out.println(dataMsg);
		send(dataMsg);
	}


	@Override
	public void onClose(int code, String reason, boolean remote) {
		logger.info("######### Okex连接断开，重连中 ######## 断开原因：" + reason);
		new Thread() {
			@Override
			public void run() {
				try {
					Thread.sleep(Const.RE_CONN_TIME);
					new OkexReceiver(session, new URI(Const.OKEX.WS_URL), okex_pairs).connect();
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}.start();
	}

	@Override
	public synchronized void onMessage(String message) {
//		logger.info("okex行情更新>>>" + message);
		Result ret = new Result();
		String json = Utils.regex("\\[(.*)\\]", message);
		JSONObject json_object = JSONObject.parseObject(json);
		String channel = (String) json_object.get("channel");
		// 取channel参数中的币种名称，ok_sub_futureusd_ltc_ticker_this_week
		String coin = Utils.regex("spot_(.+)_ticker", channel);
		// System.out.println("coin>>>"+coin);
		String pair = coin;
		List<MarketData> ms = new ArrayList<MarketData>();
		JSONObject data_object = json_object.getJSONObject("data");
		MarketData data = JSONObject.parseObject(data_object.toJSONString(), MarketData.class);
		if (data != null) {
			data.setPair(pair);
			data.setExchange(Const.OKEX.EXCHANGE);
			data.setTimestamp(System.currentTimeMillis());
			ms.add(data);
			ret.setCode(Const.SUC);
			ret.setData(ms);

			try {
				//usdt 币对的最新交易价格存到缓存（资产情况统计时 使用）
				if (pair.toUpperCase().contains("USDT")
						|| pair.toUpperCase().contains("USD")) {
					EhcacheUtils.getInstance().putCache(Const.OKEX.EXCHANGE + "#" + pair,data.getLast());
				}
			}catch (Exception e){
				logger.info("OKEX  usdt 币对的最新交易价格存到缓存 失败");
			}

			if (session != null)
				// 推到btc-web主题
				try {
					synchronized (session) {
						session.send(Const.OKEX.MARKET_TOPIC, JSONObject.toJSONString(ret).getBytes());
					}
				} catch (Exception e) {
					logger.error("", e);
				}

		}
	}

	@Override
	public void onMessage(ByteBuffer message) {
		logger.info("received ByteBuffer");
	}

	@Override
	public void onError(Exception ex) {
		logger.info("an error occurred:" + ex);
	}

	public static void main(String[] args) throws Exception {
		new OkexReceiver(new URI("wss://real.okex.com:10440/websocket/okexapi")).connect();
	}
}